#![allow(non_snake_case)]
use play_api::*;
use reqwest::Client;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use std::{collections::HashSet, time::Duration};
use tokio::process::{Child, Command};
pub struct Aria2Downloader {
    rpc_url: String,
    rpc_token: String,
    gids: Vec<String>,
}

#[derive(Debug, thiserror::Error)]
pub enum Aria2Error {
    #[error("RPC error: {0}")]
    RPCError(#[from] reqwest::Error),
    #[error("Service error: {0}")]
    ServiceError(#[from] std::io::Error),
}
fn default_client() -> reqwest::Result<Client> {
    reqwest::ClientBuilder::new()
        .timeout(Duration::from_secs(10))
        .build()
}
impl Downloader for Aria2Downloader {
    type Error = Aria2Error;
    async fn add(&mut self, task: &[DownloadTask]) -> Result<(), Self::Error> {
        let tasks: Vec<RpcRequest> = task
            .iter()
            .map(|t| RpcRequest {
                jsonrpc: "2.0".to_string(),
                id: "".to_string(),
                method: "aria2.addUri".to_string(),
                params: vec![
                    json!(format!("token:{}", self.rpc_token)),
                    json!([t.url]),
                    json!({
                        "out": t.path,
                        "continue": true,
                        "max-connection-per-server": 16,
                    }),
                ],
            })
            .collect();
        let client = default_client()?;
        let resp = client
            .post(&self.rpc_url)
            .header("Content-Type", "application/json")
            .json(&tasks)
            .send()
            .await?;
        let resp: Vec<RpcResponse> = resp.json().await?;
        for r in resp {
            if let RpcResult::String(gid) = r.result {
                self.gids.push(gid);
            }
        }
        Ok(())
    }

    async fn pause(&self) -> Result<(), Self::Error> {
        let tasks: Vec<RpcRequest> = self
            .gids
            .iter()
            .map(|g| RpcRequest {
                jsonrpc: "2.0".to_string(),
                id: "".to_string(),
                method: "aria2.pause".to_string(),
                params: vec![json!(format!("token:{}", self.rpc_token)), json!(g)],
            })
            .collect();
        let client = default_client()?;
        let _resp = client
            .post(&self.rpc_url)
            .header("Content-Type", "application/json")
            .json(&tasks)
            .send()
            .await?;
        Ok(())
    }

    async fn resume(&self) -> Result<(), Self::Error> {
        let tasks: Vec<RpcRequest> = self
            .gids
            .iter()
            .map(|g| RpcRequest {
                jsonrpc: "2.0".to_string(),
                id: "".to_string(),
                method: "aria2.unpause".to_string(),
                params: vec![json!(format!("token:{}", self.rpc_token)), json!(g)],
            })
            .collect();
        let client = default_client()?;
        let _resp = client
            .post(&self.rpc_url)
            .header("Content-Type", "application/json")
            .json(&tasks)
            .send()
            .await?;
        Ok(())
    }

    async fn cancel(&mut self) -> Result<(), Self::Error> {
        let tasks: Vec<RpcRequest> = self
            .gids
            .iter()
            .map(|g| RpcRequest {
                jsonrpc: "2.0".to_string(),
                id: "".to_string(),
                method: "aria2.remove".to_string(),
                params: vec![json!(format!("token:{}", self.rpc_token)), json!(g)],
            })
            .collect();
        let client = default_client()?;
        let resp = client
            .post(&self.rpc_url)
            .header("Content-Type", "application/json")
            .json(&tasks)
            .send()
            .await?;
        let resp: Vec<RpcResponse> = resp.json().await?;
        let to_remove: HashSet<String> = resp
            .iter()
            .map(|r| {
                if let RpcResult::TaskInfo(r) = &r.result {
                    r.status.clone()
                } else {
                    unreachable!()
                }
            })
            .collect();
        self.gids.retain(|g| !to_remove.contains(g));
        Ok(())
    }

    async fn get_status(
        &self,
    ) -> Result<(DownloaderStatus, Vec<play_api::TaskStatus>), Self::Error> {
        let tasks: Vec<_> = self
            .gids
            .iter()
            .map(|g| RpcRequest {
                jsonrpc: "2.0".to_string(),
                id: "".to_string(),
                method: "aria2.tellActive".to_string(),
                params: vec![
                    json!(format!("token:{}", self.rpc_token)),
                    json!(g),
                    json!(["totalLength", "completedLength", "downloadSpeed", "status"]),
                ],
            })
            .collect();
        let client = default_client()?;
        let resp = client
            .post(&self.rpc_url)
            .header("Content-Type", "application/json")
            .json(&tasks)
            .send()
            .await?;
        let resp: Vec<RpcResponse> = resp.json().await?;
        let tasks_status: Vec<_> = resp
            .iter()
            .map(|r| {
                if let RpcResult::TaskInfo(r) = &r.result {
                    TaskStatus {
                        total_size: r.totalLength,
                        downloaded_size: r.completedLength,
                        speed: r.downloadSpeed,
                        state: r.status.clone(),
                    }
                } else {
                    unreachable!()
                }
            })
            .collect();
        let body = RpcRequest {
            jsonrpc: "2.0".to_string(),
            id: "".to_string(),
            method: "aria2.getGlobalStat".to_string(),
            params: vec![json!(format!("token:{}", self.rpc_token))],
        };
        let resp = client
            .post(&self.rpc_url)
            .header("Content-Type", "application/json")
            .json(&body)
            .send()
            .await?;
        let resp: RpcResponse = resp.json().await?;
        if let RpcResult::GlobalStatus(r) = resp.result {
            Ok((
                DownloaderStatus {
                    speed: r.downloadSpeed,
                    active: r.numActive,
                    waiting: r.numWaiting,
                },
                tasks_status,
            ))
        } else {
            unreachable!()
        }
    }
}
#[derive(Debug, Serialize, Deserialize)]
struct RpcRequest {
    jsonrpc: String,
    id: String,
    method: String,
    params: Vec<Value>,
}
#[derive(Debug, Serialize, Deserialize)]
struct RpcResponse {
    jsonrpc: String,
    id: String,
    result: RpcResult,
}
#[derive(Debug, Serialize, Deserialize)]
#[serde(untagged)]
enum RpcResult {
    String(String),
    TaskInfo(TaskInfo),
    GlobalStatus(GlobalStatus),
}
#[derive(Debug, Serialize, Deserialize)]
struct GlobalStatus {
    downloadSpeed: u64,
    numActive: u64,
    numWaiting: u64,
}
#[derive(Debug, Serialize, Deserialize)]
struct TaskInfo {
    totalLength: u64,
    completedLength: u64,
    downloadSpeed: u64,
    status: String,
}
impl Aria2Downloader {
    pub fn new(rpc_url: &str, rpc_token: &str) -> Self {
        Self {
            rpc_url: rpc_url.to_string(),
            rpc_token: rpc_token.to_string(),
            gids: Vec::new(),
        }
    }
    pub fn init(&mut self) -> std::io::Result<Child> {
        let pid = std::process::id();
        let child = Command::new("aria2c")
            .arg("--enable-rpc")
            .arg("--rpc-listen-port=6800")
            .arg("--rpc-secret=token")
            .arg("--stop-with-process")
            .arg(pid.to_string())
            .spawn()?;

        Ok(child)
    }
}
